package com.example.kafka.service;

import com.example.kafka.entity.Order;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.persistence.EntityManager;
import java.util.concurrent.TimeUnit;

@Service
@RequiredArgsConstructor
public class IdempotentMessageService {
    private final Logger log = LoggerFactory.getLogger(getClass());


    private final EntityManager entityManager;
    private final RedisTemplate<String, String> redisTemplate;

    /**
     * 用例1：基于数据库唯一索引的幂等处理（订单创建）
     * 通过订单号唯一索引确保订单不会重复创建
     */
    @KafkaListener(topics = "order-create", groupId = "order-group")
    @Transactional
    public void handleOrderCreate(String orderNo) {
        try {
            // 检查订单是否已存在
            Order existingOrder = entityManager
                .createQuery("SELECT o FROM Order o WHERE o.orderNo = :orderNo", Order.class)
                .setParameter("orderNo", orderNo)
                .getResultList()
                .stream()
                .findFirst()
                .orElse(null);

            if (existingOrder != null) {
                log.info("订单已存在，跳过处理: {}", orderNo);
                return;
            }

            // 创建新订单
            Order order = new Order();
            order.setOrderNo(orderNo);
            order.setStatus("CREATED");
            entityManager.persist(order);

            log.info("订单创建成功: {}", orderNo);
        } catch (Exception e) {
            log.error("订单创建失败: {}", orderNo, e);
            throw e;
        }
    }

    /**
     * 用例2：基于Redis的幂等处理（积分发放）
     * 使用Redis的SETNX命令实现幂等性
     */
    @KafkaListener(topics = "points-award", groupId = "points-group")
    public void handlePointsAward(String message) {
        String key = "points:award:" + message;
        Boolean isFirst = redisTemplate.opsForValue().setIfAbsent(key, "1", 24, TimeUnit.HOURS);

        if (Boolean.FALSE.equals(isFirst)) {
            log.info("积分已发放，跳过处理: {}", message);
            return;
        }

        try {
            // 执行积分发放逻辑
            log.info("积分发放成功: {}", message);
        } catch (Exception e) {
            // 发生异常时删除Redis键，允许重试
            redisTemplate.delete(key);
            log.error("积分发放失败: {}", message, e);
            throw e;
        }
    }

    /**
     * 用例3：基于业务状态的幂等处理（支付状态更新）
     * 通过检查当前状态确保状态转换的幂等性
     */
    @KafkaListener(topics = "payment-status", groupId = "payment-group")
    @Transactional
    public void handlePaymentStatus(String orderNo) {
        Order order = entityManager
            .createQuery("SELECT o FROM Order o WHERE o.orderNo = :orderNo", Order.class)
            .setParameter("orderNo", orderNo)
            .getResultList()
            .stream()
            .findFirst()
            .orElseThrow(() -> new RuntimeException("订单不存在: " + orderNo));

        // 检查当前状态
        if ("PAID".equals(order.getStatus())) {
            log.info("订单已支付，跳过处理: {}", orderNo);
            return;
        }

        // 更新支付状态
        order.setStatus("PAID");
        entityManager.merge(order);
        log.info("支付状态更新成功: {}", orderNo);
    }

    /**
     * 用例4：基于消息ID的幂等处理（通知发送）
     * 使用消息ID作为幂等键
     */
    @KafkaListener(topics = "notification", groupId = "notification-group")
    public void handleNotification(String messageId) {
        String key = "notification:" + messageId;
        Boolean isFirst = redisTemplate.opsForValue().setIfAbsent(key, "1", 7, TimeUnit.DAYS);

        if (Boolean.FALSE.equals(isFirst)) {
            log.info("通知已发送，跳过处理: {}", messageId);
            return;
        }

        try {
            // 发送通知逻辑
            log.info("通知发送成功: {}", messageId);
        } catch (Exception e) {
            redisTemplate.delete(key);
            log.error("通知发送失败: {}", messageId, e);
            throw e;
        }
    }

    /**
     * 用例5：基于分布式锁的幂等处理（库存扣减）
     * 使用Redis分布式锁确保并发安全
     */
    @KafkaListener(topics = "inventory-deduct", groupId = "inventory-group")
    public void handleInventoryDeduct(String productId) {
        String lockKey = "inventory:lock:" + productId;
        String processKey = "inventory:process:" + productId;

        // 获取分布式锁
        Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS);
        if (Boolean.FALSE.equals(locked)) {
            log.info("获取锁失败，跳过处理: {}", productId);
            return;
        }

        try {
            // 检查是否已处理
            Boolean isProcessed = redisTemplate.opsForValue().setIfAbsent(processKey, "1", 24, TimeUnit.HOURS);
            if (Boolean.FALSE.equals(isProcessed)) {
                log.info("库存已扣减，跳过处理: {}", productId);
                return;
            }

            // 执行库存扣减逻辑
            log.info("库存扣减成功: {}", productId);
        } catch (Exception e) {
            log.error("库存扣减失败: {}", productId, e);
            throw e;
        } finally {
            // 释放分布式锁
            redisTemplate.delete(lockKey);
        }
    }
}
